package custom;

import beans.SensorReading;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * 自定义时间戳分配器，继承有界无序时间戳分配器
 *
 * @author lvbingbing
 * @date 2022-01-01 22:31
 */
public class MyTimestampExtractor extends BoundedOutOfOrdernessTimestampExtractor<SensorReading> {

    /**
     * @param maxOutOfOrderness 最大乱序程度
     */
    public MyTimestampExtractor(Time maxOutOfOrderness) {
        super(maxOutOfOrderness);
    }

    /**
     * 从给定元素中提取时间戳
     *
     * @param sensorReading <br>
     * @return 时间戳
     */
    @Override
    public long extractTimestamp(SensorReading sensorReading) {
        return sensorReading.getTimestamp() * 1000L;
    }
}
